package services

import (
	"context"
	"encoding/json"
	"log"
	"net/http"
	"time"

	"github.com/gin-gonic/gin"
	amqp "github.com/rabbitmq/amqp091-go"
)

func failOnError(err error, msg string) {
	if err != nil {
		log.Panicf("%s: %s", msg, err)
	}
}

func CV_Calculate(insertId string, resultChan chan map[string]interface{}) {
	sendToRabbit(insertId)
	listenRes(insertId, resultChan)
}

// 将消息发送给rabbitmq
func sendToRabbit(insertId string) {
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	failOnError(err, "Failed to connect to RabbitMQ")
	defer conn.Close()

	ch, err := conn.Channel()
	failOnError(err, "Failed to open a channel")
	defer ch.Close()
	// 声明一个队列用于接收响应
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	log.Println("this is the messgae", insertId)
	err = ch.PublishWithContext(ctx,
		"",          // exchange
		"rpc_queue", // routing key
		false,       // mandatory
		false,       // immediate
		amqp.Publishing{
			ContentType:   "text/plain",
			CorrelationId: insertId,
			ReplyTo:       "rpc_resp",
			Body:          []byte(insertId),
		})
	failOnError(err, "Failed to publish a message")
}

func listenRes(insertId string, resultChan chan map[string]interface{}) {
	log.Println("listenRes")
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	failOnError(err, "Failed to connect to RabbitMQ")
	defer conn.Close()

	ch, err := conn.Channel()
	failOnError(err, "Failed to open a channel")
	defer ch.Close()

	q, err := ch.QueueDeclare(
		"rpc_resp", // name
		false,      // durable
		false,      // delete when unused
		false,      // exclusive
		false,      // noWait
		nil,        // arguments
	)
	failOnError(err, "Failed to declare a queue")

	msgs, err := ch.Consume(
		q.Name, // queue
		"",     // consumer
		false,  // auto-ack
		false,  // exclusive
		false,  // no-local
		false,  // no-wait
		nil,    // args
	)
	if err != nil {
		resultChan <- map[string]interface{}{
			"status": http.StatusInternalServerError,
			"data":   gin.H{"error": "Failed to consume messages"},
		}
		return
	}

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	// use ctx to controll the goroutine
	for {
		select {
		case <-ctx.Done():
			log.Println("Context canceled, stopping listener")
			resultChan <- map[string]interface{}{
				"code": 5001,
				"data": "RPC Request timed out",
			}
			return
		case d := <-msgs:
			{
				if insertId == d.CorrelationId {
					var responseData map[string]interface{}
					err := json.Unmarshal(d.Body, &responseData)
					if err != nil {
						log.Fatalf("Failed to unmarshal JSON: %v", err)
					}

					// 检查 msg 字段是否存在且为 "Success"
					msg, ok := responseData["msg"].(string)
					if !ok || msg != "Success" {
						resultChan <- map[string]interface{}{
							"code": 5001,
							"data": msg,
						}
						d.Nack(false, false)
						return
					}

					// 如果 msg 字段为 "Success"，发送正确的响应
					resultChan <- map[string]interface{}{
						"code": 10000,
						"data": responseData,
					}
					d.Ack(false)
					return
				}
				d.Nack(false, false)
			}
		}
	}
}
